package teacher;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.*;
import java.util.*;

// 合并电影文件和评分文件
public class DataAdd {

    public static class DataAddMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

        // 用于存储电影信息的HashMap
        Map<String, String> map = new HashMap<>();

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            // 读取电影信息文件，将文件内容存储到HashMap集合中
            InputStreamReader inputStreamReader = new InputStreamReader(new FileInputStream(Config.MOVIE_PATH)/*, "GBK"*/);
            BufferedReader br = new BufferedReader(inputStreamReader);
            String line;
            while ((line = br.readLine()) != null) {
                System.out.println(line); // 可选：打印读取的行（用于调试）
                String[] split = line.split("::");
                map.put(split[0], split[1] + "::" + split[2]); // 存储电影ID、名称和类型
            }
            br.close(); // 关闭BufferedReader
        }

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] split = line.split("::");
            // 获取评分文件中的电影ID，并通过HashMap获取对应的电影信息，然后拼接成新的行
            line += ("::" + map.get(split[1]));
            context.write(new Text(line), NullWritable.get());
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration config = new Configuration();
        FileSystem fs = FileSystem.get(config);

        Job job = Job.getInstance(config);
        job.setJarByClass(DataAdd.class);
        job.setMapperClass(DataAddMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        // 不设置Reducer，因为只需要Map阶段来合并数据
        job.setNumReduceTasks(0);

        Path inputPath = new Path(Config.RATE_PATH);
        Path outputPath = new Path(Config.TEMP_PATH);

        if (fs.exists(outputPath)) {
            fs.delete(outputPath, true);
        }

        FileInputFormat.setInputPaths(job, inputPath);
        FileOutputFormat.setOutputPath(job, outputPath);

        boolean isDone = job.waitForCompletion(true);
        System.out.println(isDone ? "执行成功" : "执行失败");
    }
}